package com.bub.pay.mq.consumer.config;

import com.aliyun.openservices.ons.api.ExpressionType;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.bub.pay.core.constant.TopicCloseOrderNotifyTagConstant;
import com.bub.pay.mq.config.MqProperties;
import com.bub.pay.mq.config.TopsProperties;
import com.bub.pay.mq.consumer.listener.PayNotifyListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Configuration
public class MqConsumerConfig {

    @Resource
    private MqProperties mqProperties;

    @Resource
    private TopsProperties topsProperties;

    @Bean(name = "consumer", initMethod = "start", destroyMethod = "shutdown")
    public ConsumerBean getConsumer() {
        ConsumerBean consumer = new ConsumerBean();
        Properties properties = new Properties();
        properties.setProperty("GROUP_ID", mqProperties.getGroupId());
        properties.setProperty("AccessKey", mqProperties.getAccessKey());
        properties.setProperty("SecretKey", mqProperties.getSecretKey());
        properties.setProperty("NAMESRV_ADDR", mqProperties.getNamesrvAddr());
        properties.setProperty("ConsumeThreadNums", mqProperties.getConsumeThreadNums());
        consumer.setProperties(properties);
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
        subscriptionTable.put(subPayNotifyMsg(), getPayNotifyListener());
        consumer.setSubscriptionTable(subscriptionTable);
        consumer.start();
        return consumer;
    }

    @Bean("subMerchantDifference")
    public Subscription subMerchantDifference() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopMerchantDifference());
        subscription.setExpression("*");
        return subscription;
    }

    @Bean("subPayNotifyMsg")
    public Subscription subPayNotifyMsg() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopPayNotifyMsg());
        subscription.setExpression("*");
        return subscription;
    }

    @Bean("subPlatformDataCount")
    public Subscription subPlatformDataCount() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopPlatformDataCount());
        subscription.setExpression("*");
        return subscription;
    }

    @Bean("subRemitNotifyMsg")
    public Subscription subRemitNotifyMsg() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopRemitNotifyMsg());
        subscription.setExpression("*");
        return subscription;
    }

    @Bean("subNotifyDownStream")
    public Subscription subNotifyDownStream() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopNotifyDownStream());
        subscription.setExpression("*");
        return subscription;
    }

    @Bean("subRemitCloseOrderNotify")
    public Subscription subRemitCloseOrderNotifyMsg() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopCloseOrderNotify());
        subscription.setExpression(TopicCloseOrderNotifyTagConstant.REMIT);
        subscription.setType(ExpressionType.TAG.name());
        return subscription;
    }

    @Bean("subPayCloseOrderNotify")
    public Subscription subPayCloseOrderNotifyMsg() {
        Subscription subscription = new Subscription();
        subscription.setTopic(topsProperties.getTopCloseOrderNotify());
        subscription.setExpression("*");
        subscription.setType(ExpressionType.TAG.name());
        return subscription;
    }

    @Bean("payNotifyListener")
    public PayNotifyListener getPayNotifyListener() {
        return new PayNotifyListener();
    }


//	@Bean("notifyDownStreamListener")
//	public NotifyDownStreamListener getNotifyDownStreamListener() {
//		return new NotifyDownStreamListener();
//	}

}
